/*
 * @(#)ProcessorThread.java 1.0 2014/02/14
 *
 * Copyright (c) 2014 SinoSoft Refactored by filon51.
 * All rights reserved.
 *
 */
package com.sinosoft.cloud.dp.thread;

//import EDU.oswego.cs.dl.util.concurrent.BoundedLinkedQueue;

import org.apache.log4j.Logger;
import org.apache.log4j.NDC;

import java.util.HashMap;
import java.util.concurrent.LinkedBlockingDeque;

/**
 * 处理线程
 *
 * @author filon51
 * @since 2014-02-14
 */
public class ProcessorThread extends Thread {
	private BusinessService m_businessService = null;
	private LinkedBlockingDeque m_readDataQueue = null;
	private LinkedBlockingDeque m_writeDataQueue = null;
	private Logger m_log = Logger.getLogger(ProcessorThread.class);

	public ProcessorThread(BusinessService businessService,
						   LinkedBlockingDeque readDataQueue, LinkedBlockingDeque writeDataQueue) {
		m_businessService = businessService;
		m_readDataQueue = readDataQueue;
		m_writeDataQueue = writeDataQueue;
	}

	public void run() {
		HashMap hashElement = null;

		NDC.push(this.getName());
		m_log.debug("ProcessorThread " + this.getName() + " start");

		while (true) {
			try {
				hashElement = (HashMap) m_readDataQueue.take();
			} catch (Exception ex) {
				m_log.error("m_readDataQueue.take()", ex);
				ex.printStackTrace();
			}

			if (hashElement == MultiThreadExecutor.getOBJ_READ_GUARD()) {
				// 已经处理完所有要处理的数据了
				try {
					// 放入监视对象，告诉数据写入进程该数据处理线程已经结束。
					m_writeDataQueue.put(MultiThreadExecutor
							.getOBJ_WRITE_GUARD());
					m_log.debug("put write guard object");
				} catch (Exception ex) {
					m_log.error("m_writeDataQueue.put()", ex);
					ex.printStackTrace();
				}
				break;
			} else {
				try {
					if(m_businessService.getExceptionFlag()) {//出现异常
						m_log.error("==============other dp occurs Exception!");
						m_writeDataQueue.put(MultiThreadExecutor.getOBJ_WRITE_GUARD());
						break;
					} else {
						try {
							// 如果是用的本地连接池的话，可能会抛出连接池的错误。本地连接池的同步程度不够好。
							hashElement = m_businessService
									.processOneElement(hashElement);
							if(hashElement==null){
								m_log.error("服务实现类的processOneElement()方法出现异常，线程被迫终止，请检查");
								m_businessService.setExceptionFlag(true);
								m_writeDataQueue.put(MultiThreadExecutor.getOBJ_EXCEPTION_GUARD());
								break;
							}
							m_writeDataQueue.put(hashElement);
						}catch(Throwable ex) {
//						}catch(Exception ex) {
							m_log.error("==============ProcessorThread 调用 processOneElement 存在异常!"+ex);
							ex.printStackTrace();
							m_businessService.setExceptionFlag(true);
							m_writeDataQueue.put(MultiThreadExecutor.getOBJ_EXCEPTION_GUARD());
							break;
						}
					}
				} catch (Exception ex) {
					m_log.error("m_writeDataQueue.put()", ex);
					ex.printStackTrace();
				}
			}
		}

		m_log.debug("ProcessorThread end");
		NDC.pop();
	}
}
